<?php


namespace app\common\lib\rabbitMQ;

use app\common\service\Redis;
use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;


/**
 * Class RabbitMQ
 * @package app\common\service
 * 不同交换器
 * fanout 广播交换
 * topic 通配交换
 * direct 直接交换
 * headers
 * 是否持久化
 * 是否重发 `requeue`
 * 是否公平调度 `basic_qos` `prefetch_count = 1` 在处理和确认以前的消息之前，不要向消费者发送新消息。相反，它将发送给下一个仍然不忙的消费者。
 */
class RabbitMQ
{
    /**
     * @throws \Exception
     *
     * @Author: Li Guo Xin <guoxinlee129@gmail.com>
     * @Date: 2020/5/8
     * @Time: 9:53
     */
    public static function send(): void
    {
        $routeKey     = 'temp_key';
        $exchangeType = 'direct';
        $exchangeName = 'temp_logs';
        $queueName    = 'temp_queue_name';

        $connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');

        /**
         * @$channel_id 信道id
         */
        $channel = $connection->channel();

        /**
         * 创建交换器
         */
        $channel->exchange_declare($exchangeName, $exchangeType, false, true, false);


        #异步回调消息确认:
        //设置一个处理程序，以AMQPMessage作为第一个参数来调用服务器确认的任何消息。
        $channel->set_ack_handler(
            function (AMQPMessage $message) {
                echo "Message acked with content ".$message->body.PHP_EOL;
            }
        );
        //设置一个处理程序，以AMQPMessage作为第一个参数来调用服务器未收到的任何消息。
        $channel->set_nack_handler(
            function (AMQPMessage $message) {
                echo "Message nacked with content ".$message->body.PHP_EOL;
            }
        );

        //开启消息确认
        $channel->confirm_select();


        /**
         * 声明队列
         * 死信队列：被拒绝了消息、消息已经过期、新增消息消费超过最大重试次数后，自动进入死信队列的功能
         * @param  string  $queue  队列名称
         * @param  bool  $passive
         * @param  bool  $durable  持久化队列，相关信息和属性 会一直保留在磁盘
         * @param  bool  $exclusive  true,则只能有一个消费者独享消费
         * @param  bool  $auto_delete  如果消费者断开了消费请求，则自动删除这个message
         * @param  bool  $nowait
         */
        $channel->queue_declare($queueName, false, true, false, false);


        /**
         * 投递信息
         * @param  AMQPMessage  $msg  消息内容
         * @param  string  $exchange  交换器
         * @param  string  $routing_key  路由键
         * @param  bool  $mandatory  是否开启
         * @param  bool  $immediate  即时操作
         * @param  int|null  $ticket  门票
         */
        for ($i = 0; $i < 10000000; $i++) {
            //if($i %10 === 0){
            //    sleep(1);
            //}
            $msg = new AMQPMessage($i, [
                'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT
            ]);
            $channel->basic_publish($msg, $exchangeName, $routeKey, true, '');
        }
        echo " [x] Sent 'Hello World!'\n";

        $channel->close();
        $connection->close();
    }

    /**
     * @throws \ErrorException
     * @throws \Exception
     *
     * @Author: Li Guo Xin <guoxinlee129@gmail.com>
     * @Date: 2020/5/8
     * @Time: 10:31
     */
    public static function receive(): void
    {
        $connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');

        //声明一个信道
        $channel = $connection->channel();
        //if ($channelId = Redis::getRedis()->sRandMember('rabbitmq:channelid')) {
        //    $channel = $connection->channel($channelId);
        //} else {
        //    $channel = $connection->channel();
        //    Redis::getRedis()->sAdd('rabbitmq:channelid', $channel->getChannelId());
        //}

        $routeKey     = 'temp_key';
        $exchangeType = 'direct';
        $exchangeName = 'temp_logs';
        $queueName    = 'temp_queue_name';

        //创建叫logs的交换器
        $channel->exchange_declare($exchangeName, $exchangeType, false, true, false);

        //声明队列
        $channel->queue_declare($queueName, false, true, false, false);

        //绑定队列
        $channel->queue_bind($queueName, $exchangeName, $routeKey);

        echo " [*] Waiting for messages. To exit press CTRL+C\n";

        //回调，业务逻辑
        $callback = function ($msg) {
            echo ' [x] Received ', $msg->body, "\n";
            $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);

            #实现request-response模式
            //$properties = \AMQPBasicProperties::class;
        };

        /**
         * 指定 QoS
         *
         * @param  int  $prefetch_size  预取大小
         * @param  int  $prefetch_count  预取数量
         * @param  bool  $a_global  是否全局
         */
        $channel->basic_qos(null, 1, null);

        //启动队列消费
        $channel->basic_consume($queueName, '', false, false, false, false, $callback);

        //如果正在邮寄中，则等待
        while ($channel->is_consuming()) {
            $channel->wait();
        }

        //返回成功标识

        $channel->close();
        $connection->close();
    }
}